package springboot.kafka.util;//package com.cn.util;

import com.fasterxml.jackson.annotation.ObjectIdGenerator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import springboot.kafka.config.JedisPoolFactory;

import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 基于Redis自增策略，生成全局唯一ID
 */
@Component
public class IDGenerator {

    private static Map<String, Queue<Long>> queues = new ConcurrentHashMap<>();
    private static long section = 1; //按照时间间隔增加
    private static final String PREFIX = "ids_";

    public static String next(String idkey) {
        return _get(idkey);
    }

    private static String _get(String idkey) {
        Queue<Long> queue = getQueue(idkey);
        //此处队列肯定是存在的
        Long poll = queue.poll();
        if (poll != null) {
            return String.valueOf(poll);
        } else {
            synchronized (queue) {
                //用完了，此时加锁 去refill数据
                refill(queue, idkey);
            }
            return String.valueOf(queue.poll());
        }
    }

    private static void refill(Queue<Long> queue, String idkey) {
        Jedis connection = JedisPoolFactory.getConnection();
        Long end = connection.incrBy(PREFIX + idkey, section);
        Long start = end - section;
        initQueue(queue, start, end);
        connection.close();
    }

    private static void initQueue(Queue<Long> queue, Long start, Long end) {
        for (long i = start; i < end; i++) {
            queue.add(i);
        }
    }

    private static Queue<Long> getQueue(String idkey) {
        if (queues.containsKey(idkey)) {
            return queues.get(idkey);
        } else {
            synchronized (IDGenerator.class) {
                Queue<Long> queue = new ConcurrentLinkedQueue<>();
                queues.put(idkey, queue);
                return queue;
            }
        }
    }
}
